package com.gitee.ixtf.rsocket.broker

import cn.hutool.log.Log
import com.google.common.collect.Maps
import com.google.inject.Singleton
import io.cloudevents.CloudEvent
import io.rsocket.RSocket
import io.rsocket.util.RSocketProxy
import io.vertx.kotlin.core.json.jsonObjectOf
import java.security.Principal
import reactor.core.scheduler.Schedulers

private val CLIENT_MAP by lazy { Maps.newConcurrentMap<Principal, ClientRSocket>() }

@Singleton
class ClientManager {
  private val log by lazy { Log.get(this::class.java) }

  fun add(principal: Principal, rsocket: RSocket, cloudEvent: CloudEvent) {
    synchronized(CLIENT_MAP) {
      CLIENT_MAP[principal]?.dispose()
      CLIENT_MAP[principal] =
          ClientRSocket(id = cloudEvent.id, principal = principal, rsocket = rsocket)
      rsocket
          .onClose()
          .publishOn(Schedulers.boundedElastic())
          .doAfterTerminate { log.debug("client[${principal.name}] disconnected") }
          .subscribe()
    }
    log.debug("client[${principal.name}] connected")
  }

  fun list() = CLIENT_MAP.values.map { it.toJson() }
}

private class ClientRSocket(private val principal: Principal, val id: String, rsocket: RSocket) :
    RSocketProxy(rsocket) {
  fun toJson() = jsonObjectOf("id" to id)
}
